package com.kafka.kafkatest.util;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.google.protobuf.Api;
import com.kafka.kafkatest.entity.AigcIntoEntity;
import com.kafka.kafkatest.entity.MsgLog;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;

import java.time.LocalDateTime;
import java.util.Optional;

@Component
@Slf4j
public class KafkaConsumer {
/*
//    @KafkaListener(topics = KafkaProducer.TOPIC_TEST, groupId = KafkaProducer.TOPIC_GROUP)
    @KafkaListener( topics = KafkaProducer.TOPIC_TEST, idIsGroup = false,groupId = KafkaProducer.TOPIC_GROUP,containerFactory = "myContainerFactory")
    public void topicTest( String msg) throws InterruptedException {
//        Optional<?> message = Optional.ofNullable(record.value());
//        Optional<?> key =  Optional.ofNullable(record.key());
//        if (message.isPresent()) { // 包含非空值，则执行
//            String msg= (String) message.get();
//            AigcIntoEntity aigcIntoEntity= JSON.parseObject(msg,AigcIntoEntity.class);
            log.info(Thread.currentThread().getName()+"=========topic_test 消费了： Topic:" + ",Message:" + msg+",key:"+msg);
            Thread.sleep(4000);
            System.out.println(1/0);
//            ack.acknowledge(); // 确认成功消费一个消息
//        }
    }*/

    // 消费监听
    @KafkaListener(topics = "aigc")
    @Transactional(rollbackFor = Exception.class)
    public void onMessage1(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) throws InterruptedException {
        Optional<?> message = Optional.ofNullable(record.value());
        String msgId = record.key().toString();
//        try{
        if (message.isPresent()) {
            String msg = (String) message.get();
            log.info("aigc-开始消费=======>消息id:{},消息体{}", msgId, message);
            Thread.sleep(4000);
//                System.out.println(1 / 0);
            ack.acknowledge(); // 确认成功消费一个消息
            log.info("aigc-开始消费=======>消息msgId:{} 消费成功", msgId);
        }
//        }catch (Exception e){
//            ack.nack(2,2000); // 确认成功消费一个消息
//            e.printStackTrace();
//        }
    }



    // 消费监听
    @KafkaListener(topics = "agi")
    @Transactional(rollbackFor = Exception.class)
    public void onMessage2(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) throws InterruptedException {
        Optional<?> message = Optional.ofNullable(record.value());
        String msgId = record.key().toString();
//        try{
        if (message.isPresent()) {
            String msg = (String) message.get();
            log.info("agi-开始消费=======>消息id:{},消息体{}", msgId, message);
            Thread.sleep(4000);
//                System.out.println(1 / 0);
            ack.acknowledge(); // 确认成功消费一个消息
            log.info("agi-开始消费=======>消息msgId:{} 消费成功", msgId);
        }
//        }catch (Exception e){
//            ack.nack(2,2000); // 确认成功消费一个消息
//            e.printStackTrace();
//        }
    }
//    }
/*
        @KafkaListener(id = "my-kafka-consumer",
                idIsGroup = false, topics = KafkaProducer.TOPIC_TEST,
                containerFactory = "myContainerFactory")
        public void listen(String message) throws InterruptedException {
//            Thread.sleep(5000);
            log.info("消息消费成功,消息内容:{}", message);
        }*/


}
